package cn.doitedu.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job03_OrderDaySettlement {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        TableConfig config = tenv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true); // 开启窗口提前触发器
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "1 s");  // 每隔1s周期性触发


        // 创建cdc映射表，映射业务库中的  oms_order 表
        tenv.executeSql(
                " create table  oms_order_mysql(             "+
                        " id bigint,                         "+
                        " total_amount decimal(10,2),        "+
                        " pay_amount decimal(10,2),          "+
                        " status int,                        "+
                        " create_time timestamp(3),          "+   // 订单创建时间
                        " payment_time timestamp(3),          "+  // 支付时间
                        " delivery_time timestamp(3),         "+  // 发货时间
                        " receive_time timestamp(3),          "+  // 确认收货的时间
                        " modify_time timestamp(3),          "+
                        " rt as  modify_time,                "+
                        " watermark for rt as rt,            "+
                        " primary key (id) not enforced      "+
                        " ) with (                           "+
                        "     'connector' = 'mysql-cdc',     "+
                        "     'hostname' = 'doitedu',        "+
                        "     'port' = '3306',               "+
                        "     'username' = 'root',           "+
                        "     'password' = 'root',           "+
                        "     'database-name' = 'realtimedw',  "+
                        "     'table-name' = 'oms_order'      "+
                        " )                                    "
        );


        tenv.executeSql(
                        " select                                                                                                                      "+
                        "      'day_settelment' as key,                                                                                               "+
                        "      tumble_start(rt,interval '24' hour) as window_start,                                                                   "+
                        "      tumble_end(rt,interval '24' hour) as window_end,                                                                       "+
                        "      count(id) filter(where date_format(create_time,'yyyy-MM-dd')=CURRENT_DATE) AS `订单总数`,                                "+
                        "      sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd')=CURRENT_DATE) as `订单总额`,                        "+
                        "      sum(pay_amount) filter(where date_format(create_time,'yyyy-MM-dd')=CURRENT_DATE) as `应付总额`,                          "+
                        "                                                                                                                              "+
                        "      count(id) filter(where date_format(create_time,'yyyy-MM-dd')=CURRENT_DATE and status=0) AS `待支付订单总数`,             "+
                        "      sum(total_amount) filter(where date_format(create_time,'yyyy-MM-dd')=CURRENT_DATE and status=0) as `待支付订单总额`,     "+
                        "                                                                                                                            "+
                        "      count(id) filter(where date_format(payment_time,'yyyy-MM-dd')=CURRENT_DATE and status=1) AS `已支付订单总数`,            "+
                        "      sum(total_amount) filter(where date_format(payment_time,'yyyy-MM-dd')=CURRENT_DATE and status=1) as `已支付订单总额`,    "+
                        "                                                                                                                            "+
                        "      count(id) filter(where date_format(delivery_time,'yyyy-MM-dd')=CURRENT_DATE and status=2) AS `已发货订单总数`,           "+
                        "      sum(total_amount) filter(where date_format(delivery_time,'yyyy-MM-dd')=CURRENT_DATE and status=2) as `已返货订单总额`,   "+
                        "                                                                                                                            "+
                        "      count(id) filter(where date_format(receive_time,'yyyy-MM-dd')=CURRENT_DATE and status=3) AS `已完成订单总数`,            "+
                        "      sum(total_amount) filter(where date_format(receive_time,'yyyy-MM-dd')=CURRENT_DATE and status=3) as `已完成订单总额`     "+
                        "                                                                                                                             "+
                        " from oms_order_mysql                                                                                                        "+
                        " group by                                                                                                                    "+
                        "     tumble(rt,interval '24' hour)                                                                                           "
                ).print();


    }
}
